package org.kfive.nova.service;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.javaapi.producer.Producer;
import kafka.message.MessageAndMetadata;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.text.MessageFormat;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * Created by admin on 2017-04-04.
 */
public class MessageService {
    private Producer producer;
    private ConsumerConnector consumer;
    private Properties properties;
    private Properties consuerProperties;
    public MessageService(){
        properties = new Properties();
        consuerProperties = new Properties();
        properties.put("metadata.broker.list","192.168.86.22:9092");
        properties.put("serializer.class","kafka.serializer.StringEncoder");
        producer = new Producer(new ProducerConfig(properties));
        consuerProperties.put("zookeeper.connect","192.168.86.22:2181");
        consuerProperties.put("group.id","kfive");
        consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(consuerProperties));

    }

    public void sendMessage(String message){
        KeyedMessage<String,String> keyMessage = new KeyedMessage<String, String>("kfive","1",message);

        this.producer.send(keyMessage);
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        this.producer.close();
    }

    public void reciveMessage(){
        Map<String, List<KafkaStream<byte[], byte[]>>> messageStream = consumer.createMessageStreams(new HashMap<String, Integer>(){{put("kfive",1);}});
        KafkaStream<byte[],byte[]> streams = messageStream.get("kfive").get(0);
        ConsumerIterator<byte[],byte[]> iterator = streams.iterator();
        MessageAndMetadata<byte[],byte[]> messageAndMetadata = null;
        while (iterator.hasNext()){
            messageAndMetadata = iterator.next();
            System.out.println(MessageFormat.format("Receive->[ message:{0} , key:{1} , partition:{2} , offset:{3} ]", new String(messageAndMetadata.message()), new String(messageAndMetadata.key()), messageAndMetadata.partition() + "", messageAndMetadata.offset() + ""));
        }
    }

    public static void main(String[] args) {
        MessageService sender = new MessageService();
        sender.sendMessage("test1");
    }

}
